package com.lagou.boot;

import com.lagou.client.RPCConsumer;
import com.lagou.listener.ChildrenListener;
import com.lagou.service.IUserService;
import com.lagou.zookeeper.ZkClientFactory;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author
 */
public class ClientBootStrap {

    private static Map<String, RPCConsumer> rpcConsumerMap = new ConcurrentHashMap<>();

    private static CuratorFramework zkClient = ZkClientFactory.getZkClient();

    public static void main(String[] args) throws Exception {
        List<String> serverNodeList = zkClient.getChildren().forPath(ZkClientFactory.PROVIDER);

        //从zookeeper指定节点下，初始化可用的服务端集合
        for (String server : serverNodeList) {
            byte[] bytes = zkClient.getData().forPath(ZkClientFactory.PROVIDER + "/" + server);
            String addr = new String(bytes);
            RPCConsumer rpcConsumer = createRpcConsumer(addr);
            rpcConsumerMap.put(addr, rpcConsumer);
        }

        //建立节点监听，如果发生节点变化，更新可用服务端集合
        PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, ZkClientFactory.PROVIDER, true);
        pathChildrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        pathChildrenCache.getListenable().addListener(new ChildrenListener());

        int i = 0;
        while(true){
            //获取一个响应时间较短的，且有效的RpcConsumer
            RPCConsumer rpcConsumer = getLessCostTimeRpc();
            //当前没有可用的服务端等待重试
            if(rpcConsumer == null){
                System.out.println("==>当前没有可用的服务端，等待重试");
                Thread.sleep(2000);
                continue;
            }
            IUserService proxy = (IUserService) rpcConsumer.createProxy(IUserService.class);
            String result = proxy.sayHello("success! this is " + i + " time");
            //只有收到返回成功+1，否则进行重试
            if("success".equals(result)){
                System.out.println("编号为【" + i + "】请求的响应结果为：" + result);
                i ++;
            } else {
                System.out.println("服务器节点异常，重新发送编号为【" + i + "】的请求");
            }
        }
    }

    private static RPCConsumer createRpcConsumer(String addr) throws Exception {
        String[] split = addr.split(":");
        if(split == null || split.length != 2){
            throw new Exception("服务端节点数据异常");
        }
        String host = split[0];
        int port = Integer.parseInt(split[1]);
        RPCConsumer rpcConsumer = new RPCConsumer(host, port);
        return rpcConsumer;
    }

    public static void removeRpcConsumer(byte[] data){
        String addr = new String(data);
        rpcConsumerMap.remove(addr);
    }

    public static void addRpcConsumer(byte[] data) throws Exception {
        String addr = new String(data);
        rpcConsumerMap.put(addr, createRpcConsumer(addr));
    }

    private static RPCConsumer getLessCostTimeRpc() throws Exception {
        System.out.println("===开始选择上次响应时间较短的服务端===");
        //备选服务端集合
        List<RPCConsumer> rpcConsumerList = new ArrayList<>();
        long minCostTime = Long.MAX_VALUE;
        long costTime;
        for(Map.Entry<String, RPCConsumer> entry : rpcConsumerMap.entrySet()){
            List<String> childrenList = zkClient.getChildren().forPath(ZkClientFactory.COST_TIME + "/" + entry.getKey());
            //如果没有子节点，响应时间设置为0
            if(childrenList == null || childrenList.size() == 0){
                costTime = 0;
            } else {
                if(childrenList.size() != 1){
                    throw new Exception("服务端节点数据异常");
                }
                byte[] bytes = zkClient.getData().forPath(ZkClientFactory.COST_TIME +
                        "/" + entry.getKey() + "/" + childrenList.get(0));
                costTime = Long.valueOf(new String(bytes));
            }
            System.out.println(entry.getValue() + " last time cost " + costTime + " ms");
            //花的时间相同，加入备选集合
            if(costTime == minCostTime){
                rpcConsumerList.add(entry.getValue());
                //响应时间较短，替换原来的服务端
            } else if(costTime < minCostTime){
                minCostTime = costTime;
                rpcConsumerList.clear();
                rpcConsumerList.add(entry.getValue());
            }
        }

        //如果服务端的响应时间相同，随机选择一个服务端
        int random = new Random().nextInt(500) + 100;
        //没有可用的服务端
        if(rpcConsumerList.size() == 0){
            return null;
        }
        int index = random % rpcConsumerList.size();
        RPCConsumer rpcConsumer = rpcConsumerList.get(index);
        System.out.println("choose rpcConsumer is " + rpcConsumer);
        return rpcConsumer;
    }
}
